package com.mqtt;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 消息订阅者
 * 
 * @author zincredible
 * @date 2019/06/18 14:47:36
 */
public class Subscriber {

	// MQTT服务器
	private final static String HOST = "tcp://127.0.0.1:1883";
	// 客户端ID
	private final static String CLIENT_ID = "clientSubscribe";
	// 订阅topic
	private final static String SUBSCRIBE_TOPIC = "/v1/device/process";

	private final static DateFormat YMDHMS_FORMAT = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss:SSS");

	// mqtt客户端
	private MqttClient mqttClient;
	// 连接参数
	private MqttConnectOptions options;
	// 回调对象
	private MqttCallback callback;

	public Subscriber() {
		super();
		init();
	}

	private void init() {
		try {
			// 初始化连接参数
			options = getOptions();

			// 初始化回调对象
			callback = new CallBack(mqttClient, options);

			mqttClient = new MqttClient(HOST, CLIENT_ID, new MemoryPersistence());
			// 设置连接超时
			mqttClient.setTimeToWait(10000);
			// 连接
			mqttClient.connect(options);
			// 订阅
			mqttClient.subscribe(SUBSCRIBE_TOPIC, 2);

			// 回调方法
			mqttClient.setCallback(callback);
		} catch (MqttException e) {
			System.err.println("订阅异常：" + e.toString());
		}

	}

	// mqtt连接参数配置
	private MqttConnectOptions getOptions() {
		MqttConnectOptions options = new MqttConnectOptions();
		// 设置cleansession，若为true，当 disconnect 时,会移除这个 client 所有的 subscriptions.
		// 是否保留上一条记录
		options.setCleanSession(true);
		// 连接超时
		options.setConnectionTimeout(10);
		// 心跳会话时间
		options.setKeepAliveInterval(60);
		// 自动重连
		options.setAutomaticReconnect(true);
		return options;
	}

	public static class CallBack implements MqttCallbackExtended {

		private MqttClient client;
		private MqttConnectOptions options;

		public CallBack() {
		}

		public CallBack(MqttClient client, MqttConnectOptions options) {
			this.client = client;
			this.options = options;
		}

		// 方法实现说明 断线重连方法，如果是持久订阅，重连是不需要再次订阅，如果是非持久订阅，重连是需要重新订阅主题
		// 取决于options.setCleanSession(true); true为非持久订阅
		@Override
		public void connectionLost(Throwable cause) {
			System.out.println("服务掉线...");
		}

		@Override
		public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
			System.out.println(
					YMDHMS_FORMAT.format(new Date()) + " |[接收到新消息] " + new String(mqttMessage.getPayload(), "UTF-8"));
		}

		@Override
		public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

		}

		@Override
		public void connectComplete(boolean reconnect, String serverURI) {
			System.out.println("服务已连接...");
		}

	}

	public static void main(String[] args) {
		new Subscriber();
		System.out.println("启动订阅...");
	}

}
